package com.kun.pay.service.service.impl;

import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.kun.api.IPayService;
import com.kun.common.constant.ShopCode;
import com.kun.common.exception.CustomerException;
import com.kun.pay.service.mapper.TradePayMapper;
import com.kun.pay.service.service.TradeMqProducerTempService;
import com.kun.pojo.domain.TradeMqProducerTemp;
import com.kun.pojo.domain.TradePay;
import com.kun.pojo.entity.Result;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboService;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.validation.Valid;
import javax.validation.constraints.NotBlank;
import java.util.Date;

@Slf4j
@Service
@DubboService(interfaceClass = IPayService.class)
public class PayServiceImpl extends ServiceImpl<TradePayMapper, TradePay> implements IPayService {

    @Value("${rocketmq.producer.group}")
    private String groupName;
    @Value("${mq.pay.tag}")
    private String tag;
    @Value("${mq.pay.tag}")
    private String topic;

    @Autowired
    private TradeMqProducerTempService tradeMqProducerTempService;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    /**
     * 创建支付订单
     */
    @Override
    public Result createPayment(@Valid TradePay tradePay) {
        // 1. 判断订单支付状态
        TradePay pay = lambdaQuery().eq(TradePay::getOrderId, tradePay.getOrderId())
                .eq(TradePay::getIsPaid, ShopCode.SHOP_PAYMENT_IS_PAID.getCode()).last("LIMIT 1").one();
        if (pay != null) throw new CustomerException(ShopCode.SHOP_PAYMENT_IS_PAID);
        // 2. 设置订单状态为未支付
        tradePay.setIsPaid(ShopCode.SHOP_ORDER_PAY_STATUS_NO_PAY.getCode());
        // 3， 保存支付订单
        save(tradePay);
        return Result.success(ShopCode.SHOP_SUCCESS.getMessage());
    }

    /**
     * 回调支付结果
     */
    @Override
    public Result callbackPayment(TradePay tradePay) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        log.info("订单支付回调");
        // 1. 判断用户支付状态
        if (ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY.getCode().equals(tradePay.getIsPaid())) {
            // 2. 更新支付订单为已支付
            TradePay pay = lambdaQuery().eq(TradePay::getPayId, tradePay.getPayId()).last("LIMIT 1").one();
            if (pay == null) throw new CustomerException(ShopCode.SHOP_PAYMENT_NOT_FOUND);
            boolean isPayFlag = lambdaUpdate().set(TradePay::getIsPaid, ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY.getCode())
                    .eq(TradePay::getPayId, tradePay.getPayId()).update();
            log.info("支付订单状态改为成功");
            if (isPayFlag) {
                // 3. 创建支付成功消息
                String jsonString = JSON.toJSONString(tradePay);
                TradeMqProducerTemp tradeMqProducerTemp = new TradeMqProducerTemp(groupName, topic, tag,
                        tradePay.getPayId().toString(), jsonString, new Date());
                String idStr = IdWorker.getIdStr();
                tradeMqProducerTemp.setId(idStr);
                // 4. 将消息持久化数据库
                tradeMqProducerTempService.save(tradeMqProducerTemp);
                log.info("持久化支付成功消息到数据库");
                // 5. 发送消息到 MQ--线程池优化
                threadPoolTaskExecutor.execute(() -> {
                    try {
                        SendResult sendResult = sendMessage(topic, tag, tradePay.getPayId().toString(), jsonString);
                        if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
                            log.info("消息发送成功");
                            // 6. 等待发送结果，如果MQ接收到消息，删除数据库发送成功的消息
                            tradeMqProducerTempService.removeById(idStr);
                            log.info("消息删除成功");
                        }
                    } catch (MQBrokerException | RemotingException | InterruptedException | MQClientException e) {
                        e.printStackTrace();
                    }
                });
            }
            return Result.success(ShopCode.SHOP_SUCCESS.getMessage());
        } else throw new CustomerException(ShopCode.SHOP_PAYMENT_PAY_ERROR);
    }

    /**
     * 发送消息
     */
    private SendResult sendMessage(@NotBlank(message = "请求参数不存在") String topic, String tag,
                                   String key, @NotBlank(message = "请求参数不存在") String body)
            throws MQBrokerException, RemotingException, InterruptedException, MQClientException {

        Message message = new Message(topic, tag, key, body.getBytes());
        return rocketMQTemplate.getProducer().send(message);

    }
}




